package com.atguigu.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HbaseDML {


    public static void main(String[] args) throws IOException {


            putData("gmall","customer","1001","info","name","l444");
            putData("gmall","customer","1001","info","age","33");
            putData("gmall","customer","1002","info","name","jack");
            putData("gmall","customer","1002","info","age","44");
            putData("gmall","customer","1003","info","name","rose");
            putData("gmall","customer","1003","info","age","11");
        putData("gmall","customer","1004","info","name","tom");
        putData("gmall","customer","1004","info","age","11");

        System.out.println(" 写入执行完毕");
//            putData("gmall","customer","1001","info","sex","男");
//            putData("gmall","customer","1001","info","address","北京");
//            putData("gmall","customer","1001","info","phone","13800000000");

        String value = getCell("gmall", "customer", "1001", "info", "name");
        System.out.println("value = " + value);


        List<String> list = scanData("gmall", "customer", "1001", "1004~", "info", "name");
        for (String name : list) {
            System.out.println("name = " + name);
        }

        delete("gmall","customer","1001","info","name");

        getCell("gmall", "customer", "1001", "info", "name");
        deleteAll("gmall", "customer", "1001" );
    }


    /**
     * 将数据放入HBase表中
     *
     * @param namespace 表所属的命名空间
     * @param tableNameStr 表的名称
     * @param rowKey 行键
     * @param family 列族
     * @param qualifier 列限定符
     * @param value 要存储的值
     * @throws IOException 如果发生I/O错误
     */
    public static void putData(String namespace,String tableNameStr,String rowKey,String family,String qualifier,String value) throws IOException {
        // 1获取HBase连接
        Connection connection = HbaseConnect.getConnection();

        // 2创建TableName对象，指定命名空间和表名
        TableName tableNameObj  = TableName.valueOf(namespace, tableNameStr);

        // 3通过连接获取表对象
        Table table = connection.getTable(tableNameObj);

        // 4 组织put
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));

        //5 执行put
        table.put(put);


        //   关闭表对象，释放资源
        table.close();


    }


    public static String getCell(String namespace,String tableNameStr,String rowKey,String family,String qualifier) throws IOException {

        //1 获得链接
        Connection connection = HbaseConnect.getConnection();
        //2 创建TableName对象
        TableName tableNameObj = TableName.valueOf(namespace, tableNameStr);
        //3 获得会话对象Table
        Table table = connection.getTable(tableNameObj);
        //4 创建Get对象
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
      //  get.readAllVersions();

        //5  获得结果
        Result result = table.get(get);

        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println("value = " + value);

            return value;
        }

        return null;
    }


    public static List<String> scanData(String namespace,String tableNameStr,String startRow ,String stopRow,String family,String qualifier) throws IOException {

        //1 获得链接
        Connection connection = HbaseConnect.getConnection();
        //2 创建TableName对象
        TableName tableNameObj = TableName.valueOf(namespace, tableNameStr);
        //3 获得会话对象Table
        Table table = connection.getTable(tableNameObj);

        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startRow));
        scan.withStopRow(Bytes.toBytes(stopRow));

        //如果使用单值过滤器 不要绑定扫描行
      //  scan.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));

        //增加一个列值过滤器
        ColumnValueFilter columnValueFilter = new ColumnValueFilter(
                Bytes.toBytes(family),
                Bytes.toBytes(qualifier),
                CompareOperator.EQUAL,
                Bytes.toBytes("l4")
        );
        //单值过滤器
        SingleColumnValueFilter ageFilter = new SingleColumnValueFilter(Bytes.toBytes(family),
                Bytes.toBytes("age"),
                CompareOperator.EQUAL,
                Bytes.toBytes("11")
        );

      //  scan.setFilter(columnValueFilter);
         scan.setFilter(ageFilter);

        ResultScanner scanner = table.getScanner(scan);

         List<String> arrayList = new ArrayList();
        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                System.out.println("---"+new String(CellUtil.cloneValue(cell)) + "---");
            }

        }
        return  arrayList;


    }

    public static void delete(    String namespace,String tableNameStr,String rowKey,String family,String qualifier  ) throws IOException {

        //1 获得链接
        Connection connection = HbaseConnect.getConnection();
        //2 创建TableName对象
        TableName tableNameObj = TableName.valueOf(namespace, tableNameStr);
        //3 获得会话对象Table
        Table table = connection.getTable(tableNameObj);

        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addColumn(Bytes.toBytes(family),Bytes.toBytes(qualifier));
        table.delete(delete);
        table.close();
    }

    public static void deleteAll(    String namespace,String tableNameStr,String rowKey  ) throws IOException {

        //1 获得链接
        Connection connection = HbaseConnect.getConnection();
        //2 创建TableName对象
        TableName tableNameObj = TableName.valueOf(namespace, tableNameStr);
        //3 获得会话对象Table
        Table table = connection.getTable(tableNameObj);

        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        table.close();
    }
}
